package com.easou.ps.exec;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;

/**
 * this class maintains a thread pool internally for executing task,
 * multiple thread should share a same instance of this class,
 * it is supposed to be called by the main processing thread, the calling sequence is: 
 * prepare - execTask - execTask - ... - getResults
 * 
 * @author changliu
 * @param <T>
 *
 */
public class Executor<T> {
	public static final Logger LOG = Logger.getLogger("server");
	
	private volatile long lastTime = 0;
	
	private final int interval = 1 * 60 * 1000;
	private AtomicInteger totalActiveCount = new AtomicInteger(0);
	private AtomicInteger processCount = new AtomicInteger(0);
	private static AtomicLong lastSendMailTime = new AtomicLong(0);

	private ThreadPoolExecutor exec;
	private int timeoutMills;
	private String executorName = "";
	
	/**
	 * this threadlocal holds the type and result involved in different task.
	 */
	private ThreadLocal<List<Future<T>>> futures = new ThreadLocal<List<Future<T>>>() {
		public List<Future<T>> initialValue() {
			return new ArrayList<Future<T>>();
		}
	};
	
	public Executor(ThreadPoolExecutor exec, int timeoutMills){
		this.exec = exec;
		this.timeoutMills = timeoutMills;
	}

	public Executor(String executorName, ThreadPoolExecutor exec, int timeoutMills) {
		super();
		this.executorName = executorName;
		this.exec = exec;
		this.timeoutMills = timeoutMills;
	}

	public void prepare() {
		futures.get().clear();
	}
	
	public void execTask(Callable<T> task) {
		futures.get().add(exec.submit(task));
	}
	
	public List<T> getResults() {
		return getResults(timeoutMills);
	}
	
	public List<T> getResults(int mills) {
		List<T> ret = new ArrayList<T>();
		List<Future<T>> list = futures.get();
		if (list.size() == 0) {
			return ret;
		}
		for (int i = 0; i < list.size(); i++) {
			long start = System.currentTimeMillis();
			T rs = null;
			Future<T> future = list.get(i);
			try {
				rs = future.get(mills, TimeUnit.MILLISECONDS);
				
			} catch (Exception e) {
				future.cancel(true);
				long lost = System.currentTimeMillis()-start;
				LOG.error(executorName + " Executor: main thread waiting results timeout... cost: " + lost);
			}
			ret.add(rs);
			mills -= (System.currentTimeMillis() - start);
			if (mills < 0) {
				mills = 0;
			}
		}
		
		
		int procCnt = processCount.incrementAndGet();
		totalActiveCount.addAndGet(exec.getActiveCount());
		if (System.currentTimeMillis() - lastTime > interval) {
			lastTime = System.currentTimeMillis();
			
			LOG.info(executorName + " Executor > active threads:"
					+ exec.getActiveCount() + ", current threads:"
					+ exec.getPoolSize() + ",queue size:"
					+ exec.getQueue().size() + ",avg active threads:"
					+ (((float) totalActiveCount.get()) / procCnt));
			
			processCount.set(0);
			totalActiveCount.set(0);
			
			if(exec.getQueue().remainingCapacity()<50){
				try {
					LOG.info(executorName + " Executor > quene size near full!" +
							"active threads:" + exec.getActiveCount() + ", current threads:" + exec.getPoolSize()+",queue size:"+exec.getQueue().size());
					if(System.currentTimeMillis()-lastSendMailTime.get()>0){
						lastSendMailTime.set(System.currentTimeMillis());
						String subject = executorName + " Executor quene size near full";
						String msg = "active threads:" + exec.getActiveCount() + ", current threads:" + exec.getPoolSize()+",queue size:"+exec.getQueue().size();
						LOG.info(subject +"_"+ msg);	
					}
				} catch (Exception e) {
					LOG.info(executorName + " Executor > send mail fail!", e);
				}
			}
		}
		
		return ret;
	}
}
